package com.iot.service.mqtt.handler;

import com.iot.service.mqtt.config.Constant;
import com.iot.service.mqtt.model.entity.TlSubClient;
import com.iot.service.mqtt.model.entity.TlTopic;
import com.iot.service.mqtt.model.request.TlMqttSubscribeReq;
import com.iot.service.mqtt.model.response.TlMqttSubAck;
import com.iot.service.mqtt.service.retain.IRetainService;
import com.iot.service.mqtt.service.session.ISessionService;
import com.iot.service.mqtt.service.subscripe.ISubscriptionService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
@Component
@RequiredArgsConstructor
public class TlMqttSubScribeHandler extends SimpleChannelInboundHandler<TlMqttSubscribeReq> {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttSubScribeHandler.class);

    private final ISubscriptionService subscriptionService;

    private final ISessionService sessionService;

    private final IRetainService retainService;




    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttSubscribeReq msg) throws Exception {
        logger.debug("in【SUBSCRIBE】handler");
        Channel channel = ctx.channel();
        List<TlTopic> topics =  msg.getPayload().getTopics();
        String clientId =channel.attr(AttributeKey.valueOf(Constant.CLIENT_ID)).get().toString();

         sessionService.find(clientId)
                       .flatMap(session -> {
                            Set<String> subTopics = session.getTopics();
                            Set<String> collect = topics.stream().map(TlTopic::getName).collect(Collectors.toSet());
                           //移除所有与传入主题同名的旧主题
                            collect.forEach(subTopics::remove);
                            // 添加所有新主题 （自动去重）
                            subTopics.addAll(collect);
                            return sessionService.save(session);
                       }).subscribe();
        /*
         * 需要将客户端订阅的主题进行保存起来
         * 后面根据这个主题就能够立马找到对应的客户端 key就是主题的名称 最好能够获取到这个主题的订阅的qos等级,然后获取到这个客户端通道 将消息转发给他
         */
        Flux.fromIterable(topics)
                .flatMap(topic -> {
                    TlSubClient client = new  TlSubClient (topic.getQos(),clientId,topic.getName());
                    return Mono.fromRunnable(()->
                                    logger.debug("client【{}】subscribe topic 【{}】",clientId,topic.getName()))
                            .then(retainService.find(topic.getName())
                                    .doOnNext(channel::writeAndFlush)
                                    .then(subscriptionService.subscribe(client))
                            .thenReturn(client)
                    );
                })
                .subscribe(null,error->logger.error("subscribe error",error),()->logger.debug("subscribe complete"));

        int messageId = msg.getVariableHead().getMessageId();
        TlMqttSubAck res = TlMqttSubAck.of(topics, messageId);
        channel.writeAndFlush(res);
        logger.debug("build【SUBACK】return");

    }
}
